…used var

Review changes:
- Default `client_rack` to `None` instead of `""` in Consumer, Fetcher,
  and FetchRequest so the wire encoding sends null (-1) rather than an
  empty string (0) when rack awareness is disabled, matching the Java client.
- Remove unused `rack_id` property from FetchRequest.
- Remove `metadata_max_age_ms` param from Fetcher; read the TTL for the
  preferred-replica cache directly from `client._metadata_max_age_ms`.
  This also removes `self._metadata_max_age_ms` from AIOKafkaConsumer.
- Rename `node_id` to `preferred_read_replica` in
  `_update_preferred_read_replica` for readability.
- Make `responder_node_id` a required keyword argument (no default);
  remove the conservative `None` fallback branch and its test since
  the caller always knows who responded.
- Rename local `parts`/`new_parts` to `partitions`/`new_partitions`
  in FetchRequest.build().

CI fixes:
- Add explicit type annotations (`list[tuple[int, ...]]`,
  `list[object]`) in FetchRequest.build() to satisfy mypy.

… test flakiness

- Revert rack_id parameter from `str | None = None` back to `str = ""`,
  matching the Kafka protocol definition (STRING, not NULLABLE_STRING).
- Remove `or ""` fallback in FetchRequest.build_request() since rack_id
  is now always a string.
- Add `or ""` at the call site in fetcher.py to convert None client_rack
  to empty string.
- Replace time.sleep-based TTL expiry test with direct cache manipulation
  to fix flakiness on Windows (coarse timer resolution).
- Add tests for rack version warning logging:
  - test_warning_logged_when_broker_below_v11: verifies warning is logged
    once when client_rack is set but broker doesn't support FetchRequest v11.
  - test_no_warning_when_rack_not_set: verifies no warning when client_rack
    is None.

…ead_replica

The leader-vs-follower invalidation branch is unreachable: when the
fetch was routed to the leader, the cache for that partition was
already empty by construction. Collapse to a single guarded write
and rely on TTL / transport-failure / routing-error paths for real
invalidation. Per @vmaurin's review.

Annotate `preferred_read_replica` as `int` and drop the redundant
`is not None` guard. The value always comes from FetchResponse v11+
parsing, which yields an int (`-1` when absent), so `None` is no longer
reachable.

…r, trim per-fetch overhead

Align KIP-392 preferred-replica handling with the Java consumer and cut
work done on every fetch response when the feature is inactive.

* Invalidate cached preferred replica on any non-NoError code.
  Previously only NotLeaderForPartition / UnknownTopicOrPartition and
  OffsetOutOfRange evicted the cache, so any other partition-level error
  kept routing fetches to a follower the broker had implicitly rejected
  until the TTL expired. The per-error branches are now folded into a
  single `else:` that pops `_preferred_read_replica[tp]` first and only
  then dispatches the error-specific handling (metadata refresh, offset
  reset, auth warning, fallback log). Matches Java consumer behaviour.

* Compute `rack_aware` once per response.
  KIP-392 only applies when `client_rack` is set AND the broker speaks
  FetchRequest v11+. Previously every partition in every response paid
  for a `part_data[3]` lookup and an `_update_preferred_read_replica`
  call, even though a non-rack-aware setup is guaranteed to receive -1
  and never update the cache. The per-partition update is now gated by
  a single `rack_aware` boolean computed once per response.

* Make the v<11 fallback warning truly one-shot.
  The old `client_rack and API_VERSION < 11 and not _rack_warning_logged`
  triple-check ran on every fetch response for the lifetime of the
  consumer, even though the warning itself was emitted only once. The
  check now reuses `rack_aware`, short-circuits on `bool(self._client_rack)`
  for the common (unset) case, and a comment makes the latch explicit.

tests/test_rack_awareness.py (review feedback):
- Remove unused `from __future__ import annotations`
- Remove `test_fetch_request_v9_does_not_carry_rack_id` (no requirement)
- Drop redundant `@pytest.mark.asyncio` (project uses auto mode)
- Remove unused `import pytest`
- Fix `_make_fetcher` annotation: `str` → `str | None`
- Use `mock.patch.object(asyncio, "sleep", ...)` instead of string path
- Replace `mock.patch("...create_task")` with async no-op routine

tests/test_transactional_consumer.py (Py3.14 flake fix):
Replace racy `asyncio.sleep(1)` + bare offset assertions with
`await consumer.getmany(timeout_ms=1000)` — a deterministic fetch
round-trip that guarantees the transaction control marker has been
processed and last_stable_offset/highwater are updated before asserting.
Fixes consistent `AssertionError: 0 != 3` on all Py3.14 CI jobs.

tests/test_producer.py (leader-not-found flake fix):
Warm up the topic with a separate default-timeout producer before
constructing the 200ms producer. Separates topic auto-creation from the
leader-not-found retry path the test exercises. Use `add_cleanup` to
prevent unclosed producer resource warnings on failure.